package com.sparrow.rpc.socket.handler;

import com.sparrow.rpc.manager.RpcNodeManager;
import com.sparrow.rpc.service.RpcService;
import com.sparrow.rpc.socket.msg.resp.RpcLoginResponse;
import com.sparrow.rpc.socket.msg.resp.RpcNodeBroadcastResponse;
import com.sparrow.rpc.socket.msg.resp.RpcResponse;
import com.sparrow.rpc.socket.msg.vo.RpcNodeVo;
import com.sparrow.rpc.socket.session.RpcClientSessionManager;
import com.sparrow.rpc.socket.session.RpcSession;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.Attribute;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@ChannelHandler.Sharable
public class RpcClientHandler extends ChannelInboundHandlerAdapter {

    @Autowired
    RpcClientSessionManager sessionManager;

    @Autowired
    RpcNodeManager nodeManager;

    private static RpcClientHandler INSTANCE;

    public static RpcClientHandler getInstance() {
        return INSTANCE;
    }

    public RpcClientHandler() {
        INSTANCE = this;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        RpcSession session = new RpcSession();
        session.setChannel(ctx.channel());
        ctx.channel().attr(RpcSession.SESSION).set(session);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Attribute<RpcSession> attr = ctx.channel().attr(RpcSession.SESSION);
        RpcSession rpcSession = attr.get();
        if (rpcSession != null) {
            sessionManager.close(rpcSession);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
       if (msg instanceof RpcLoginResponse) {
           RpcSession session = ctx.channel().attr(RpcSession.SESSION).get();
           RpcLoginResponse response = (RpcLoginResponse) msg;
           sessionManager.bind(session, response.getTargetNodeId());
           nodeManager.onRpcNodeLoginResp(session, response);
        } else if (msg instanceof RpcNodeBroadcastResponse) {
           RpcNodeBroadcastResponse response = (RpcNodeBroadcastResponse) msg;
           for (RpcNodeVo nodeVo : response.getNodeList()) {
               nodeManager.registerNode(nodeVo.getId(), nodeVo.getIp(), nodeVo.getPort());
           }
       } else if (msg instanceof RpcResponse) {
           RpcResponse response = (RpcResponse) msg;
           RpcService.getInstance().response(response);
       }
    }
}
